package com.zp.datax.core.util;

import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.Objects;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created by bruce on 2019/5/9.
 */
@Slf4j
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public enum ThreadPoolFactory {

    INSTANCE;

    // 根据处理器个数定义线程个数
    public final static int CORE_POOL_SIZE;
    // 根据处理器个数定义最大线程个数
    public final static int MAXIMUM_POOL_SIZE;


    static {
        final int processNum = Runtime.getRuntime().availableProcessors();
        final int defaultCorePoolSize = Math.min(processNum, 4) * 5;
        final int defaultMaximumPoolSize = defaultCorePoolSize * 2;
        final String corePoolSizeStr = System.getProperty("pool.thread.corePoolSize", Objects.toString(defaultCorePoolSize));
        final String maximumPoolSizeStr = System.getProperty("pool.thread.maximumPoolSize", Objects.toString(defaultMaximumPoolSize));

        CORE_POOL_SIZE = Integer.parseInt(corePoolSizeStr);
        MAXIMUM_POOL_SIZE = Integer.parseInt(maximumPoolSizeStr);
    }

    /**
     * 创建线程池
     */
    private volatile static ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
            CORE_POOL_SIZE
            , MAXIMUM_POOL_SIZE
            , 60L
            , TimeUnit.SECONDS
            , new ArrayBlockingQueue<>(100)
            , Executors.defaultThreadFactory()
            , new ThreadPoolExecutor.DiscardPolicy()
    );

    /**
     * 获取线程池
     *
     * @return
     */
    public ThreadPoolExecutor getThreadPool() {
        return threadPool;
    }

    /**
     * 关闭线程池
     */
    public void shutdown() {
        if (!threadPool.isShutdown()) {
            threadPool.shutdown();
            log.info(">>>>>>>>>>> 关闭线程池.");
        }

    }

    /**
     * 线程池监控
     */
    public void monitor() {
        //任务总数量
        final long taskCount = threadPool.getTaskCount();
        //当前活动线程数量
        final int activeCount = threadPool.getActiveCount();
        //当前完成任务数量
        final long completedTaskCount = threadPool.getCompletedTaskCount();
        //线程池中维持的最小线程数量
        final int corePoolSize = threadPool.getCorePoolSize();
        //线程池中维持的最大线程数量
        final int maximumPoolSize = threadPool.getMaximumPoolSize();

        //活动线程数大于0时才输出，避免过多输出无效日志
        if (activeCount > 0) {

            log.info("**********************************************************************************************************************");
            log.info("                        >>>>>>>>>> 线程池中维持的最小线程数：{}", corePoolSize);
            log.info("                        >>>>>>>>>> 线程池中维持的最大线程数：{}", maximumPoolSize);
            log.info("                        >>>>>>>>>> 线程池中任务总数：{}", taskCount);
            log.info("                        >>>>>>>>>> 线程池中活动线程数：{}", activeCount);
            log.info("                        >>>>>>>>>> 线程池中已完成任务数：{}", completedTaskCount);
            log.info("**********************************************************************************************************************");

        }


    }

}
